package com.rem.flink.flink1Base;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 批处理
 *
 * @author Rem
 * @date 2022-09-26
 */

public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        //2从文件中读取数据 按行读取
        DataSource<String> lineDataSource = environment.readTextFile("/Users/rem/mydata/project/rem_project/flink/input");
        //3.转换数据格式 到元组
        FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDataSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {

            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        //4.按照word 进行分组 根据Tuple2里第一个字段
        UnsortedGrouping<Tuple2<String, Long>> tuple2UnsortedGrouping = wordAndOne.groupBy(0);

        //5.分组内聚合统计 根据Tuple2里第二个字段
        AggregateOperator<Tuple2<String, Long>> sum = tuple2UnsortedGrouping.sum(1);
        //6.打印结果
        sum.print();

    }
}
